#include "logger.h"
#include "mysqlmgr.h"

#include "mysql/include/errmsg.h"

static MysqlCmd keepalive_sqlcmd("SELECT 11;", "mysql_keepalive");

MysqlCmd::MysqlCmd(const char* sqltxt, const char* type)
: _result(false)
, _type(type)
, _sqltxt(sqltxt)
, _mysqlrowcount(0)
, _mysqlfieldcount(0)
, _mysqlaffectedrows(0)
, _mysqlres(NULL)
, _mysqlrows(NULL)
{
}

MysqlCmd::~MysqlCmd()
{
    DEL_ARRAY(_mysqlrows);

    if(NULL != _mysqlres)
    {
        mysql_free_result(_mysqlres);
    }
}

void MysqlCmd::CreateResInfo()
{
    if(NULL == _mysqlres) return;
    
    _mysqlrowcount = mysql_num_rows(_mysqlres);
    _mysqlfieldcount = mysql_num_fields(_mysqlres);
    
    if(_mysqlrowcount > 0)
    {
        _mysqlrows =  new MYSQL_ROW[_mysqlrowcount];

        for(int index = 0; index < _mysqlrowcount; ++index)
        {
            _mysqlrows[index] = mysql_fetch_row(_mysqlres);
        }
    }
}

int MysqlCmd::GetRowCount()
{
    return _mysqlrowcount;
}

int MysqlCmd::GetFieldsCount()
{
    return _mysqlfieldcount;
}

int MysqlCmd::GetAffectedRows()
{
    return _mysqlaffectedrows;
}

const char* MysqlCmd::GetFieldName(int index)
{
    if(index < 0 || index >= _mysqlfieldcount) return "";
    
    MYSQL_FIELD* field = mysql_fetch_field_direct(_mysqlres, index);

    if(NULL == field) return "";
    
    return field->name;
}

const char* MysqlCmd::GetRowData(int rowindex, int columindex)
{
    if(rowindex < 0 || rowindex >= _mysqlrowcount) return "";
    
    if(columindex < 0 || columindex >= _mysqlfieldcount) return "";

    return _mysqlrows[rowindex][columindex];
}

THREAD_HANDLE(cow_mysql_thread, param)
{
    CCowMysql* cowmysql = (CCowMysql*)param;

    cowmysql->WorkThread();

    return 0;
}

CCowMysql::CCowMysql(SqlInfo& sqlinfo, mysql_callback_func* callback)
: m_isrun(false)
, m_pmysql(NULL)
, m_excutetime(0)
, m_callback(callback)
{
    m_log = GET_LOG("_mysql");

    m_sqlinfo = sqlinfo;
}

CCowMysql::~CCowMysql()
{
    m_isrun = false;

    if(NULL != m_pmysql)
    {
        mysql_close(m_pmysql);
    }

    sleep_s(1);
}

bool CCowMysql::Start()
{
    if(m_isrun) return false;

    m_isrun = true;

    if(!m_thread.Start(cow_mysql_thread, this)) return false;

    return true;
}

bool CCowMysql::Connect()
{
    m_pmysql = mysql_init(NULL);

    if(NULL == m_pmysql)
    {
        LOG_ERROR(m_log, "failed to init [mysql:%s] error!\n", m_sqlinfo._dbname.c_str());

        return false;
    }
    
    char value = 1;
    mysql_options(m_pmysql, MYSQL_OPT_RECONNECT, (char*)&value);
    mysql_options(m_pmysql, MYSQL_OPT_READ_TIMEOUT, (char*)&value);
    mysql_options(m_pmysql, MYSQL_OPT_WRITE_TIMEOUT, (char*)&value);
    mysql_options(m_pmysql, MYSQL_SET_CHARSET_NAME, "utf8");

    if(NULL == mysql_real_connect(m_pmysql, m_sqlinfo._ip.c_str(), m_sqlinfo._username.c_str(),
        m_sqlinfo._userpwd.c_str(), m_sqlinfo._dbname.c_str(), m_sqlinfo._port, NULL, CLIENT_MULTI_STATEMENTS | CLIENT_FOUND_ROWS))
    {
        LOG_ERROR(m_log, "failed to connect to [mysql:%s] err=%s\n",m_sqlinfo._dbname.c_str(), mysql_error(m_pmysql));

        return false;
    }

    return true;
}

void CCowMysql::WorkThread()
{
    bool isconnect = true;

    while(m_isrun)
    {
        MysqlCmd* sqlcmd = NULL;

        if(isconnect)
        {
            if(Connect())
                isconnect = false;
            else
            {
                if(0 == m_sqlincmds.Pop(sqlcmd))
                {
                    if(NULL == m_callback)
                        m_sqloutcmds.Push(sqlcmd);
                    else
                        (*m_callback)(sqlcmd);
                }

                continue;
            }
        }
        
        if(0 != m_sqlincmds.Pop(sqlcmd))
        {
            if(get_tick_count() > m_excutetime + 3 * 60 * 1000)
                sqlcmd = &keepalive_sqlcmd;
            else
                continue;
        }

        if("mysql_keepalive" != sqlcmd->_type)
        {
            LOG_INFO(m_log, "%s\n", sqlcmd->_sqltxt.c_str());
        }

        m_excutetime = get_tick_count();

        if(0 != mysql_real_query(m_pmysql, sqlcmd->_sqltxt.c_str(), (unsigned long)sqlcmd->_sqltxt.size()))
        {
            if(0 != mysql_ping(m_pmysql))
            {
                mysql_close(m_pmysql);

                int reconnect_count = 3;

                while(reconnect_count--)
                {
                    if(Connect()) break;
                }

                if(-1 == reconnect_count)
                {
                    if(NULL == m_callback)
                        m_sqloutcmds.Push(sqlcmd);
                    else
                        (*m_callback)(sqlcmd);

                    isconnect = true; continue;
                }
            }

            if(0 != mysql_real_query(m_pmysql, sqlcmd->_sqltxt.c_str(), (unsigned long)sqlcmd->_sqltxt.size()))
            {
                if("mysql_keepalive" != sqlcmd->_type)
                {
                    if(NULL == m_callback)
                        m_sqloutcmds.Push(sqlcmd);
                    else
                        (*m_callback)(sqlcmd);
                }
                
                continue;
            }
        }

        if("mysql_keepalive" == sqlcmd->_type)
            continue;

        sqlcmd->_mysqlres = mysql_store_result(m_pmysql);

        sqlcmd->_mysqlaffectedrows = mysql_affected_rows(m_pmysql);

        if(NULL == sqlcmd->_mysqlres)
        {
            if(NULL == m_callback)
                m_sqloutcmds.Push(sqlcmd);
            else
                (*m_callback)(sqlcmd);
        }
        else
        {
            sqlcmd->CreateResInfo();

            sqlcmd->_result = true;

            if(NULL == m_callback)
                m_sqloutcmds.Push(sqlcmd);
            else
                (*m_callback)(sqlcmd);
        }
    }
}

void CCowMysql::SyncWork(MysqlCmd& incmd)
{
    if(NULL != m_callback) return;

    m_sqlincmds.Push(&incmd);

    MysqlCmd* outcmd = NULL;

    while(0 == m_sqloutcmds.Pop(outcmd));

    uint excutetime = get_tick_count();

    while(get_tick_count() <= excutetime + 30 * 1000 && 0 != m_sqloutcmds.Pop(outcmd));
}
